/*
 * $Id$
 * --------------------------------------------------------------------------------------
 * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
 *
 * The software in this package is published under the terms of the CPAL v1.0
 * license, a copy of which has been included with this distribution in the
 * LICENSE.txt file.
 */

package org.mule.transport.jpa;

import java.text.MessageFormat;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;

import org.apache.commons.lang.StringUtils;
import org.mule.api.MuleContext;
import org.mule.api.MuleEvent;
import org.mule.api.MuleException;
import org.mule.api.endpoint.ImmutableEndpoint;
import org.mule.api.endpoint.InboundEndpoint;
import org.mule.api.expression.ExpressionManager;
import org.mule.api.lifecycle.InitialisationException;
import org.mule.api.service.Service;
import org.mule.api.transaction.Transaction;
import org.mule.api.transaction.TransactionException;
import org.mule.transaction.TransactionCoordination;
import org.mule.transaction.XaTransaction;
import org.mule.transaction.XaTransaction.MuleXaObject;
import org.mule.transport.AbstractConnector;
import org.mule.util.TemplateParser;

/**
 * <code>JpaConnector</code> TODO document
 */
public class JpaConnector extends AbstractConnector {
	/* This constant defines the main transport protocol identifier */
	public static final String JPA = "jpa";

	public static final String PROPERTY_QUERY_KEY = "queryKey";
	public static final String PROPERTY_POLLING_FREQUENCY = "pollingFrequency";
	public static final String PROPERTY_MAX_FETCH_SIZE = "maxFetchSize";
	public static final long DEFAULT_POLLING_FREQUENCY = 1000;
	public static final long DEFAULT_MAX_FETCH_SIZE = 1000;

	private static final Pattern STATEMENT_ARGS = TemplateParser.WIGGLY_MULE_TEMPLATE_PATTERN;

	protected long pollingFrequency = 0;
	protected long maxFetchSize;
	protected String persistenceUnit = null;
	protected Map<String, String> queries;

	private EntityManagerFactory entityManagerFactory;
	private EntityManager entityManager;

	/*
	 * For general guidelines on writing transports see
	 * http://www.mulesoft.org/documentation
	 * /display/MULE3USER/Creating+Transports
	 */

	/*
	 * IMPLEMENTATION NOTE: All configuaration for the transport should be set
	 * on the Connector object, this is the object that gets configured in
	 * MuleXml
	 */

	public JpaConnector(MuleContext context) {
		super(context);
	}

	@Override
	public void doInitialise() throws InitialisationException {
		createMultipleTransactedReceivers = false;
		// emf = Persistence.createEntityManagerFactory(this.persistenceUnit);
	}

	protected org.mule.api.transport.MessageReceiver createReceiver(Service service, InboundEndpoint endpoint) throws Exception {
		Map<?, ?> props = endpoint.getProperties();
		if (props != null) {
			String tempPolling = (String) props.get(PROPERTY_POLLING_FREQUENCY);
			if (tempPolling != null) {
				pollingFrequency = Long.parseLong(tempPolling);
			}
			String tempMaxFetchSize = (String) props.get(PROPERTY_MAX_FETCH_SIZE);
			if (tempMaxFetchSize != null) {
				maxFetchSize = Long.parseLong(tempMaxFetchSize);
			}
		}
		if (pollingFrequency <= 0) {
			pollingFrequency = DEFAULT_POLLING_FREQUENCY;
		}
		if (maxFetchSize <= 0) {
			maxFetchSize = DEFAULT_MAX_FETCH_SIZE;
		}
		return getServiceDescriptor().createMessageReceiver(this, service, endpoint,
				new Object[] { new Long(pollingFrequency), new Long(maxFetchSize) });
	}

	@Override
	public void doConnect() throws Exception {
		// Optional; does not need to be implemented. Delete if not required

		/*
		 * IMPLEMENTATION NOTE: Makes a connection to the underlying resource.
		 * When connections are managed at the receiver/dispatcher level, this
		 * method may do nothing
		 */
	}

	@Override
	public void doDisconnect() throws Exception {
		// Optional; does not need to be implemented. Delete if not required

		/*
		 * IMPLEMENTATION NOTE: Disconnects any connections made in the connect
		 * method If the connect method did not do anything then this method
		 * shouldn't do anything either.
		 */
	}

	@Override
	public void doStart() throws MuleException {
		// Optional; does not need to be implemented. Delete if not required

		/*
		 * IMPLEMENTATION NOTE: If there is a single server instance or
		 * connection associated with the connector i.e. Jms Connection or Jdbc
		 * Connection, this method should put the resource in a started state
		 * here.
		 */
	}

	@Override
	public void doStop() throws MuleException {
		// Optional; does not need to be implemented. Delete if not required

		/*
		 * IMPLEMENTATION NOTE: Should put any associated resources into a
		 * stopped state. Mule will automatically call the stop() method.
		 */
	}

	@Override
	public void doDispose() {
		if (entityManager != null) {
			entityManager.close();
		}
		if (entityManagerFactory != null) {
			entityManagerFactory.close();
		}
	}

	public String getProtocol() {
		return JPA;
	}

	public EntityManager getEntityManager() {
		Transaction tx = TransactionCoordination.getInstance().getTransaction();
		if (tx != null) {
			if (tx.hasResource(entityManager)) {
				logger.debug("Retrieving session from current transaction");
				Object r = tx.getResource(entityManager);
				if (r instanceof MuleXaSession)
					return ((MuleXaSession) r).impl;
				else
					return (EntityManager) r;
			}
		}
		if (entityManager == null || entityManager.isOpen() == false) {
			entityManager = entityManagerFactory.createEntityManager();
		}
		if (tx != null) {
			logger.debug("Binding session to current transaction");
			try {
				Object r;
				if (tx instanceof XaTransaction)
					r = new MuleXaSession(entityManager);
				else
					r = entityManager;

				tx.bindResource(entityManager, r);
			} catch (TransactionException e) {
				throw new RuntimeException("Could not bind connection to current transaction", e);
			}
		}
		return entityManager;
	}

	public void closeEntityManager() {
		if (entityManager != null && entityManager.isOpen()) {
			entityManager.close();
			entityManager = null;
		}
	}

	/**
	 * @return the queries
	 */
	public Map<String, String> getQueries() {
		return queries;
	}

	/**
	 * @param queries
	 *            the queries to set
	 */
	public void setQueries(Map<String, String> queries) {
		this.queries = queries;
	}

	/**
	 * @return the pollingFrequency
	 */
	public long getPollingFrequency() {
		return pollingFrequency;
	}

	/**
	 * @param pollingFrequency
	 *            the pollingFrequency to set
	 */
	public void setPollingFrequency(long pollingFrequency) {
		this.pollingFrequency = pollingFrequency;
	}

	/**
	 * @return the maxFetchSize
	 */
	public long getMaxFetchSize() {
		return maxFetchSize;
	}

	/**
	 * @param maxFetchSize
	 *            the maxFetchSize to set
	 */
	public void setMaxFetchSize(long maxFetchSize) {
		this.maxFetchSize = maxFetchSize;
	}

	public EntityManagerFactory getEntityManagerFactory() {
		return entityManagerFactory;
	}

	public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
		this.entityManagerFactory = entityManagerFactory;
	}

	public String getQuery(ImmutableEndpoint endpoint, String stmtKey) {
		Object query = null;
		if (endpoint != null && endpoint.getProperties() != null) {
			Object queries = endpoint.getProperties().get("queries");
			if (queries instanceof Map) {
				query = ((Map<?, ?>) queries).get(stmtKey);
			}
		}
		if (query == null) {
			if (this.queries != null) {
				query = this.queries.get(stmtKey);
			}
		}
		return query == null ? null : query.toString();
	}

	public String getStatement(ImmutableEndpoint endpoint) {
		String writeStmt = endpoint.getEndpointURI().getAddress();
		String str;
		if ((str = getQuery(endpoint, writeStmt)) != null) {
			writeStmt = str;
		}
		writeStmt = StringUtils.trimToEmpty(writeStmt);
		if (StringUtils.isBlank(writeStmt)) {
			throw new IllegalArgumentException("Missing statement");
		}
		return writeStmt;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public String parseStatement(String stmt, List params) {
		if (stmt == null) {
			return stmt;
		}
		Matcher m = STATEMENT_ARGS.matcher(stmt);
		StringBuffer sb = new StringBuffer(200);
		while (m.find()) {
			String key = m.group();
			m.appendReplacement(sb, "?");
			// Special legacy handling for #[payload]
			if (key.equals("#[payload]")) {
				// MULE-3597
				logger.error("invalid expression template #[payload]. It should be replaced with #[payload:] to conform with the correct expression syntax. Mule has replaced this for you, but may not in future versions.");
				key = "#[payload:]";
			}
			params.add(key);
		}
		m.appendTail(sb);
		return sb.toString();
	}

	@SuppressWarnings("rawtypes")
	public Object[] getParams(ImmutableEndpoint endpoint, List paramNames, MuleEvent event, String query) throws Exception {
		Object[] params = new Object[paramNames.size()];
		for (int i = 0; i < paramNames.size(); i++) {
			String param = (String) paramNames.get(i);
			Object value = getParamValue(endpoint, event, param);

			params[i] = value;
		}
		return params;
	}

	protected Object getParamValue(ImmutableEndpoint endpoint, MuleEvent event, String param) {
		Object value = null;
		// If we find a value and it happens to be null, that is acceptable
		boolean foundValue = false;
		boolean validExpression = muleContext.getExpressionManager().isValidExpression(param);

		// There must be an expression namespace to use the ExpressionEvaluator
		// i.e. header:type
		if (event.getMessage() != null && validExpression) {
			value = muleContext.getExpressionManager().evaluate(param, event);
			foundValue = value != null;
		}
		if (!foundValue) {
			String name = getNameFromParam(param);
			// MULE-3597
			if (!validExpression) {
				logger.warn(MessageFormat.format("Config is using the legacy param format {0} (no evaluator defined)."
						+ " This expression can be replaced with {1}header:{2}{3}", param, ExpressionManager.DEFAULT_EXPRESSION_PREFIX, name,
						ExpressionManager.DEFAULT_EXPRESSION_POSTFIX));
			}
			value = endpoint.getProperty(name);
		}
		return value;
	}

	protected String getNameFromParam(String param) {
		return param.substring(2, param.length() - 1);
	}

	private class MuleXaSession implements MuleXaObject {
		private EntityManager impl;

		MuleXaSession(EntityManager em) {
			this.impl = em;
		}

		public void close() throws Exception {
			if (impl != null && impl.isOpen()) {
				impl.close();
			}
		}

		public boolean enlist() throws TransactionException {
			return true;
		}

		public boolean delist() throws Exception {
			return true;
		}

		public Object getTargetObject() {
			return impl;
		}

		public boolean isReuseObject() {
			return false;
		}

		public void setReuseObject(boolean reuseObject) {
		}

	}
}
